/*
 * Copyright (c) 2012, marco.tamburelli@gmail.com
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met: 
 * 
 * 1. Redistributions of source code must retain the above copyright notice, this
 *    list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution. 
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
 * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
package org.bmi.gwt.mi.server.core.queue;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.bmi.gwt.mi.server.core.event.HandlersManager;
import org.bmi.gwt.mi.server.core.event.HandlersManager.MiEventExecutor;
import org.bmi.gwt.mi.shared.communication.attribute.AttributeValue;
import org.bmi.gwt.mi.shared.communication.call.Call;

/**
 * A queue is a tool that receives messages and wait that some client consumes
 * them..
 * 
 * The class defines a blocking queue suitable for pushing any number of
 * elements and to consume all inserted elements. The polling operation requires
 * a deferred method which will be invoked only when there are available
 * elements.
 * 
 * @author marco.tamburelli@gmail.com
 */
public class MiQueue
{
	static protected Logger logger = Logger.getLogger(MiQueue.class);

	private final HandlersManager handlersManager;

	private HashMap<String, Object> session;
	private List<CallEnvelope> callEnvelopes;

	private QueueState state;

	private long queueExpireTime;

	private class CallEnvelope
	{
		Call call;
		long callExpireTime;

		CallEnvelope(Call call, long publishingMillisecs)
		{
			this.call = call;
			this.callExpireTime = publishingMillisecs + call.getTimeToLive();

			/*
			 * In case the new incoming call has an expiration which could
			 * extends the queue expiration date, it will be applied.
			 */
			if (MiQueue.this.queueExpireTime < this.callExpireTime) MiQueue.this.queueExpireTime = this.callExpireTime;
		}
	}

	MiQueue(HandlersManager handlersManager, QueueState state)
	{
		logger.info("[MiQueue] - opening queue: " + state.getQueueName());

		this.handlersManager = handlersManager;

		this.session = new HashMap<String, Object>();
		this.callEnvelopes = new ArrayList<CallEnvelope>(50);

		this.state = state;
	}

	MiQueue(HandlersManager handlersManager, String queueName)
	{
		logger.info("[MiQueue] - opening queue: " + queueName + " (withouth owner)");

		this.handlersManager = handlersManager;

		this.session = new HashMap<String, Object>();
		this.callEnvelopes = new ArrayList<CallEnvelope>(50);

		createDummyDescriptor(queueName);
	}

	private void consumeCalls(List<CallEnvelope> envelopes, long currentTime)
	{
		List<Call> callList = new ArrayList<Call>();

		for (CallEnvelope envelope : envelopes)
			if (envelope.callExpireTime >= currentTime)
			{
				Call call = envelope.call;

				final MiEventExecutor miHM = this.handlersManager.prepareMiEvents(this.state.getQueueName(), call);
				miHM.executeOnBefores();

				if (miHM.canConsume())
				{
					callList.add(call);

					this.handlersManager.asyncExecute(new Runnable()
					{
						@Override
						public void run()
						{
							miHM.executedOnAfters();
						}
					});
				}
			}

		if (this.state instanceof CallHandler) ((CallHandler) this.state).onCalls(this.state.getQueueName(), callList);
	}

	/**
	 * Publishes a list of callEnvelopes.
	 * 
	 * @param publishTime The time expressed in system milliseconds calculated
	 *        at the moment of publishing.
	 * @param callEnvelopes
	 */
	synchronized void publish(List<Call> calls, long publishTime)
	{
		ArrayList<CallEnvelope> envelopes = new ArrayList<CallEnvelope>(calls.size());

		for (Call call : calls)
		{
			// TODO Following line should be removed in release versions
			logger.debug("[MiQueue] - publishing:" + call.toString() + " on queue " + this.state.getQueueName());
			envelopes.add(new CallEnvelope(call, publishTime));
		}

		if (this.state.canConsumeCalls())
			consumeCalls(envelopes, publishTime);
		else
			this.callEnvelopes.addAll(envelopes);
	}

	/**
	 * Sets the queue inner state; if the state allow consuming calls (when the
	 * method {@link QueueState#canConsumeCalls()} return true) all present call
	 * is consumed.
	 * 
	 * Note that to make the consuming operation effective the state also must
	 * implements the interface {@link CallHandler}.
	 * 
	 * @param state New queue status.
	 * @param consumeTime The time expressed in system milliseconds calculated
	 *        at the moment of starting consuming session (a peer has just been
	 *        bound to this queue). Of course if the status doesn't allow the
	 *        queue to consume calls, the parameter is ignored.
	 */
	synchronized void setState(QueueState state, long consumeTime)
	{
		this.state = state;

		if (this.state.canConsumeCalls() && !this.callEnvelopes.isEmpty())
		{
			consumeCalls(this.callEnvelopes, consumeTime);
			this.callEnvelopes.clear();
		}
	}

	/**
	 * Releases the queue and makes it available for another client.
	 * 
	 * Note that if no client will propose itself as new owner, the queue will
	 * be sooner or later disposed.
	 */
	synchronized void free()
	{
		createDummyDescriptor(this.state.getQueueName());
	}

	/**
	 * returns <code>true</code> if the current user can bind the queue.
	 * 
	 * @return
	 */
	synchronized boolean isFree()
	{
		return this.state.isReplaceable();
	}

	private void createDummyDescriptor(final String queueName)
	{
		this.state = new QueueState()
		{
			@Override
			public String getQueueName()
			{
				return queueName;
			}

			@Override
			public Map<String, AttributeValue<?>> getAttributes()
			{
				/* Returns an empty map */
				return new HashMap<String, AttributeValue<?>>();
			}

			@Override
			public String getOwnerCredentials()
			{
				return null;
			}

			@Override
			public boolean isReplaceable()
			{
				return true;
			}

			@Override
			public boolean canConsumeCalls()
			{
				return false;
			}
		};
	}

	/**
	 * Returns the current queue expiration time in milliseconds.
	 * 
	 * @return
	 */
	long getExpireTime()
	{
		return this.queueExpireTime;
	}

	public String getName()
	{
		return this.state.getQueueName();
	}

	QueueState getState()
	{
		return this.state;
	}

	/**
	 * Returns <code>true</code> in case the provided credentials are recognized
	 * from the owner of this queue.
	 * 
	 * @param credentials User credentials string.
	 * @return
	 */
	boolean isOwner(String credentials)
	{
		return this.state != null && this.state.getOwnerCredentials().equals(credentials);
	}

	/**
	 * Returns a queue attribute value.
	 * 
	 * @param key The attribute key.
	 * @return
	 */
	AttributeValue<?> getAttribute(String key)
	{
		if (this.state == null || this.state.getAttributes() == null) return null;

		return this.state.getAttributes().get(key);
	}

	/**
	 * Sets a queue attributes.
	 * 
	 * @param key The attribute key.
	 * @param value The attribute value
	 */
	void setAttribute(String key, AttributeValue<?> value)
	{
		if (this.state == null) return;

		Map<String, AttributeValue<?>> attributes = this.state.getAttributes();

		if (attributes == null) return;

		attributes.put(key, value);
	}

	/**
	 * Removes an attribute and returns its value.
	 * 
	 * @param key The attribute key.
	 * @return
	 */
	AttributeValue<?> removeAttribute(String key)
	{
		if (this.state == null) return null;

		Map<String, AttributeValue<?>> attributes = this.state.getAttributes();

		if (attributes == null) return null;

		return attributes.remove(key);
	}

	/**
	 * Provides the queue session. A session for a queue is a useful place where
	 * to store objects that could be needed in any part of server client
	 * concerning queue and bound peer behaviors, since the tool doesn't provide
	 * an efficient and easy way to reach the servlet session.
	 * 
	 * @return
	 */
	public HashMap<String, Object> getSession()
	{
		return this.session;
	}
}
